package com.sui.bigdata.flink.sql.function.table;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.TableFunction;

public class SplitJsonUdtf extends TableFunction<Tuple2<String, String>> {

    // 可选， open方法可不编写。若编写，需要添加声明'import org.apache.flink.table.functions.FunctionContext;'。
    @Override
    public void open(FunctionContext context) {
    }

    public void eval(String str) {
        JSONArray jsonArray = null;
        try {
            jsonArray = JSONArray.parseArray(str);
        } catch (Exception e) {
        }
        if (jsonArray != null) {
            for (int i = 0; i < jsonArray.size(); i++) {
                JSONObject jsonObject = jsonArray.getJSONObject(i);
                Tuple2<String, String> tuple = new Tuple2<>(jsonObject.getString("name"), jsonObject.getString("packName"));
                collect(tuple);
            }
        }
    }

    // 可选，close方法可不编写。
    @Override
    public void close() {
    }
}